package cn.spark.study.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.Seconds

class TransformBlacklist {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
        .setMaster("local[2]")
        .setAppName("TransformBlacklist")
    
    val ssc = new StreamingContext(conf,Seconds(3))
      
    val blacklist = Array(("Tom",true),("Jack",true))
    val blacklistRDD = ssc.sparkContext.parallelize(blacklist, 2)
    
    val clickLogDSteam = ssc.socketTextStream("spark1", 9999)
    val pairRDD = clickLogDSteam.map { log => (log.split(" ")(0),log) }
    
    val validClickLogRDD = pairRDD.transform(pair => {
       val leftJoinRDD = pair.leftOuterJoin(blacklistRDD)
       val filterRDD = leftJoinRDD.filter(rdd => {
         if(rdd._2._2.isDefined && rdd._2._2.get){
           false
         }else{
         true
         }
       })
       filterRDD.map(_._2._1)
    })
    
    validClickLogRDD.print()
    ssc.start()
    ssc.awaitTermination()
  }
}